#define UDP_C
/*----------------------------------------------*
 * include files            	 ������ͷ�ļ�   *
 *----------------------------------------------*/
#if defined(OS_ANDROID)||defined(OS_LINUX_SOC)
#include <sys/types.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <unistd.h>
#include <err.h>
#include<sys/prctl.h>
#elif defined(OS_WIN32)
#include <winsock2.h>
#include <ws2tcpip.h>
#endif

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <fcntl.h>
#include <errno.h>

//user header
#include <libmscommon/mscommon.h>
#include <libmscommon/msthread.h>
#include <libmscommon/msstring.h>
#include <libmscommon/msnetwork.h>
#include <libmscommon/msepoll.h>
#include <libmscommon/mssysctl.h>
#include <libmslog/mslog.h>
#include "msprotocol.h"
#include "igmp.h"

#define FLAG "UDP"

#define MTF_UDPBUF				0
#define UDP_TX_BUFSIZE			4194304
#define UDP_RX_BUFSIZE			4194304
#define UDP_HEADER_SIZE 		8
#define UDPLITE_SEND_CSCOV		10
#define UDPLITE_RECV_CSCOV		11

typedef struct UDPContext {
	ms_bool is_multicast;		//udp use
	int  udplite_coverage; 	//udplite use
	ms_s32 msepoll_fd;
	ms_u32 msepoll_opt;
} UDPContext;

/*----------------------------------------------*
 * constants  ����                              *
 *----------------------------------------------*/

/*----------------------------------------------*
 * routines' implementations   ������ʵ��       *
 *----------------------------------------------*/
static void udp_optinit(URLContext *s)
{
	OPTVAL_INIT("client_prot",s->opt.udp.client_prot, 0, -1);
	OPTVAL_INIT("block",s->opt.udp.block, 0, 0);
	OPTVAL_INIT("rw_timeout",s->opt.udp.rw_timeout, 0, msnet_default_timeout_rw);	//Note:client's timeout cannot less than rw_timeout
	OPTVAL_INIT("reuse_socket",s->opt.udp.reuse_socket, 0, 1);
	OPTVAL_INIT("fifo_size",s->opt.udp.fifo_size, 0, (7*4096));
} 
static ms_s08 udp_innerapi_getHostname(URLContext *s)
{
#if ENABLE_IPV6
	MSIPlist msiplist;
	ms_memset(&msiplist,0,sizeof(MSIPlist));
	ms_u08 useip_index=0;
	msnet_api_parseDomainName (s->url_info.hostname, ms_null,&msiplist) ;
	if( (msiplist.totalnum>0)&&(useip_index<msiplist.totalnum) ){
		ms_debug("===========url %s,[%d]ipaddr,flag_ipv6:%s,%d",s->url,useip_index,msiplist.ipaddr[useip_index],msiplist.flag_ipv6[useip_index]);
		s->url_info.flag_ipv6=msiplist.flag_ipv6[useip_index];
		ms_sprintfs(s->url_info.hostname, "%s", msiplist.ipaddr[useip_index]);
		useip_index++;
	}else{
		ms_errRet(-1, "used all ip to try,faild,goto end");
	}
#endif
	return 0;
}
static int udp_innerapi_createSocket(URLContext *s)
{
	UDPContext * udp_ctt=s->priv_data;
	int sct_af=0;
	if( ENABLE_IPV6 && ( s->url_info.flag_ipv6 || (!(s->flags&FLAG_WRITE)  && (!udp_ctt->is_multicast))  )	){
		sct_af=AF_INET6;
		s->url_info.flag_useipv6=ms_true;
	}else{
		sct_af=AF_INET;
		s->url_info.flag_useipv6=ms_false;
	}
	if (ms_strncmp_seq(s->url_info.ptcName,"udplite")){
#if defined(OS_ANDROID)||defined(OS_LINUX_SOC)
		s->fd =socket(sct_af, SOCK_DGRAM, IPPROTO_UDPLITE);
#else
		s->fd =socket(sct_af, SOCK_DGRAM, 0);
#endif
	}else{
		s->fd =socket(sct_af, SOCK_DGRAM, 0);
	}
	if (s->fd < 0){
		ms_errRet(-1,  "socket %s failed(%d,%s)", s->url,errno,strerror(errno));
	}
	return 0;
}
static char * udp_innerapi_getLocalIp(URLContext *s,int *ret)
{
	UDPContext * udp_ctt=s->priv_data;
	char *localhostip=ms_null;
	if( ms_true==udp_ctt->is_multicast){
		localhostip=(s->flags&FLAG_WRITE) ? s->opt.udp.ipOut : s->opt.udp.ipIn;
		if((ms_null==localhostip)){
			(*ret)=-1;
			ms_errRet(ms_null, "localhostip is null");
		}
		char *ifcname=(s->flags&FLAG_WRITE) ? s->opt.udp.ifcOut : s->opt.udp.ifcIn;
		//如果输出是组播，但未获取到输出网卡的IP
		if(ms_strncmp_saeq(localhostip, "0.0.0.0")){
			if(ms_null!=ifcname){
				//尝试进行IP地址的获取
				msnet_api_getIfaceIp(ifcname,localhostip);
				if((ms_null==localhostip)||ms_strncmp_saeq(localhostip, "0.0.0.0")){
					(*ret)=-1;
					ms_errRet(ms_null,  "localhostip(%s) is null or 0.0.0.0",s->opt.udp.ifcOut);
				}
			}else{
				(*ret)=-1;
				ms_errRet(ms_null,  "localhostip is null or 0.0.0.0");
			}
		}else{
			ms_debug("localhostip is %s",localhostip);
		}
	}

	(*ret)=0;
	return localhostip;
}
static ms_s08 udp_innerapi_setBufSize(URLContext *s)
{
	int optname=0;
	ms_string strOptname=ms_null;
	if(!(s->flags&FLAG_WRITE)	){
		optname=SO_RCVBUF;
		strOptname="SO_RCVBUF";
	}else{
		optname=SO_SNDBUF;
		strOptname="SO_SNDBUF";
	}
#if MTF_UDPBUF	
	int old_bufsize=0;
	int len=sizeof(old_bufsize);
	getsockopt(s->fd, SOL_SOCKET, optname, (char *)&old_bufsize, (socklen_t * __restrict__)&len);
	ms_verbose("[%s] %s:%d ", s->url,strOptname,old_bufsize);
	///proc/sys/net/core/rmem_default
	///proc/sys/net/core/rmem_max
	int new_bufsize=(s->flags&FLAG_WRITE) ? UDP_TX_BUFSIZE : UDP_RX_BUFSIZE;
	if(old_bufsize!=new_bufsize){
		ms_verbose("[%s]try to set %s form %d to %d", s->url,strOptname,old_bufsize,new_bufsize);
		if (setsockopt(s->fd, SOL_SOCKET, optname, (const char *)&new_bufsize, sizeof(new_bufsize)) < 0) {
			ms_close(s->fd);
			ms_errGoto(fail, "setsockopt(%s) %s(%d,%s)", s->url,strOptname, errno, strerror(errno));
		}else{
			int new_bufsize1=0;
			len=sizeof(new_bufsize1);
			getsockopt(s->fd, SOL_SOCKET, optname, (char *)&new_bufsize1, (socklen_t * __restrict__)&len);
			ms_verbose("[%s]change %s form %d to %d", s->url,strOptname,old_bufsize,new_bufsize1);
		}		
	}
#else
	ms_debug("disable  %s:MTF_UDPBUF",strOptname );
#endif		
	return 0;	
fail:
	return -1;
}
static ms_void udp_innerapi_listenAddrInit(URLContext *s,struct sockaddr_in6 *listen_addr6,struct sockaddr_in *listen_addr,char *localhostip)
{
	UDPContext * udp_ctt=s->priv_data;
	memset(listen_addr6, 0, sizeof(struct sockaddr_in6));
	listen_addr6->sin6_family = AF_INET6;
	listen_addr6->sin6_port = htons(s->url_info.port);

	memset(listen_addr, 0, sizeof(struct sockaddr_in));
	listen_addr->sin_family = AF_INET;
	listen_addr->sin_port = htons(s->url_info.port);	
	if(s->flags&FLAG_WRITE){
		if(udp_ctt->is_multicast){
			if(s->url_info.flag_useipv6){
				if(ms_null!=localhostip){
					inet_pton(AF_INET6,localhostip,&listen_addr6->sin6_addr);
				}else{
					listen_addr6->sin6_addr = in6addr_any;
				}
			}else{
				listen_addr->sin_addr.s_addr = (ms_null!=localhostip) ? inet_addr(localhostip) : htonl(INADDR_ANY);
			}
		}else{
			if(s->url_info.flag_useipv6){
				inet_pton(AF_INET6,s->url_info.hostname,&listen_addr6->sin6_addr);
			}else{
				listen_addr->sin_addr.s_addr = inet_addr(s->url_info.hostname);
			}
		}
	}else{
		//组播输入如果不绑定地址，当流是相同端口时，会报error-no 98(Address already in use)
		if(s->url_info.flag_useipv6){
			if(udp_ctt->is_multicast){
				inet_pton(AF_INET6,s->url_info.hostname,&listen_addr6->sin6_addr);
			}else{
				listen_addr6->sin6_addr = in6addr_any;
			}
		}else{
			listen_addr->sin_addr.s_addr = (udp_ctt->is_multicast) ? inet_addr(s->url_info.hostname) : htonl(INADDR_ANY);
		}
	}		

}
static ms_void udp_innerapi_setPeerInfo(URLContext *s)
{
	if( (s->flags&FLAG_READ_WRITE)){
		if(s->url_info.flag_useipv6){
			memset(&s->peer_info.sockaddrs6 ,0 ,sizeof(s->peer_info.sockaddrs6));
			s->peer_info.len=sizeof(s->peer_info.sockaddrs6);
			s->peer_info.sockaddrs6.sin6_family = AF_INET6;
			if((0<s->opt.udp.client_prot)
				&&ms_null!=s->opt.udp.client_hostname){
				s->peer_info.sockaddrs6.sin6_port = htons(s->opt.udp.client_prot);
				ms_byte client_hostname[128]={0};
				ms_strcpy(client_hostname,s->opt.udp.client_hostname);
				msnet_api_outIpv6(client_hostname);
				inet_pton(AF_INET6,(client_hostname),&s->peer_info.sockaddrs6.sin6_addr);
				s->peer_info.has_info=1;
				//ms_debug("get peer info");
			}
		}else{
			memset(&s->peer_info.sockaddrs ,0 ,sizeof(struct sockaddr_in));
			s->peer_info.len=sizeof(s->peer_info.sockaddrs);
			s->peer_info.sockaddrs.sin_family = AF_INET;
			if((0<s->opt.udp.client_prot)
				&&ms_null!=s->opt.udp.client_hostname){
				s->peer_info.sockaddrs.sin_port = htons(s->opt.udp.client_prot);
				s->peer_info.sockaddrs.sin_addr.s_addr=inet_addr(s->opt.udp.client_hostname);	
				s->peer_info.has_info=1;
				//ms_debug("get peer info");
			}
		}
	}else if( (s->flags&FLAG_WRITE)){
		if(s->url_info.flag_useipv6){
			memset(&s->peer_info.sockaddrs6 ,0 ,sizeof(s->peer_info.sockaddrs6));
			s->peer_info.len=sizeof(s->peer_info.sockaddrs6);
			s->peer_info.sockaddrs6.sin6_family = AF_INET6;
			s->peer_info.sockaddrs6.sin6_port = htons(s->url_info.port);
			inet_pton(AF_INET6,(s->url_info.hostname),&s->peer_info.sockaddrs6.sin6_addr);
		}else{
			memset(&s->peer_info.sockaddrs ,0 ,sizeof(struct sockaddr_in));
			s->peer_info.len=sizeof(s->peer_info.sockaddrs);
			s->peer_info.sockaddrs.sin_family = AF_INET;
			s->peer_info.sockaddrs.sin_port = htons(s->url_info.port);
			s->peer_info.sockaddrs.sin_addr.s_addr=inet_addr(s->url_info.hostname);
		}
		//ms_debug("get peer info");
	}	

}
static ms_s08 udp_innerapi_setIO(URLContext *s)
{
	if (0==s->opt.udp.block ){
		if (msnet_api_socketNonblock(s->fd, ms_true) < 0){
			ms_errGoto(fail, "msnet_api_socketNonblock failed");
		}
	}else{
		ms_debug("[%s]-------------blocking", s->url);
		struct timeval	timeout={1,0};
		if (setsockopt(s->fd, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout, sizeof(struct timeval)) < 0) {
			ms_close(s->fd);
			ms_errGoto(fail, "setsockopt(SO_RCVTIMEO) %s(%d,%s)", s->url, errno, strerror(errno));
		}	
		if (setsockopt(s->fd, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout, sizeof(struct timeval)) < 0) {
			ms_close(s->fd);
			ms_errGoto(fail, "setsockopt(SO_SNDTIMEO) %s(%d,%s)", s->url, errno, strerror(errno));
		}
	}
	return 0;	
fail:
	return -1;
}
static ms_s08 udp_innerapi_igmp(URLContext *s)
{
	if (setsockopt (s->fd, SOL_SOCKET, SO_REUSEADDR,(const char *) &(s->opt.udp.reuse_socket), sizeof(s->opt.udp.reuse_socket)) != 0){
		ms_close(s->fd);
		ms_errGoto(fail, "setsockopt(SO_REUSEADDR) %s(%d,%s)", s->url,errno,strerror(errno));
	}
	ms_verbose("whitelist,blacklist num---%d,%d",s->opt.udp.num_include_sources,s->opt.udp.num_exclude_sources);
	ms_s32 igmp_version=igmp_api_getVersion();
	if(s->opt.udp.num_include_sources>0){
		ms_debug("include_sources:igmp version:%d",igmp_version);
		if(3!=igmp_version){
			ms_waring("Want to use include_sources,but igmp version is %d,expect igmp v3", igmp_version);
		}
		igmp_api_joinGroupSources(s, ms_true);
	}else{
		igmp_api_join(s);
	}
	if(s->opt.udp.num_exclude_sources>0){
		ms_debug("exclude_sources:igmp version:%d",igmp_version);
		if(3!=igmp_version){
			ms_waring("Want to use exclude_sources,but igmp version is %d,expect igmp v3", igmp_version);
		}
		igmp_api_joinGroupSources(s, ms_false);
	}
	return 0;	
fail:
	return -1;
}
static ms_s08 udp_innerapi_bind(URLContext *s,char *localhostip)
{
	UDPContext * udp_ctt=s->priv_data;
	char udpio_type=0;
	if(!(s->flags&FLAG_WRITE)){
		udpio_type=1;
	}	
	if((s->flags&FLAG_WRITE || s->flags&FLAG_READ_WRITE) &&udp_ctt->is_multicast && (ms_null!=localhostip)){
		udpio_type=2;
	}
	struct in_addr if_req; 
	switch(udpio_type){
		case 1: //UDP输入，绑定IP地址和端口
			ms_verbose("Init add info");
			struct sockaddr_in6 listen_addr6;
			struct sockaddr_in listen_addr;
			udp_innerapi_listenAddrInit(s, &listen_addr6, &listen_addr, localhostip);
			ms_verbose("Bind sokect....");
			int ret_bind;
			if(s->url_info.flag_useipv6){
				ret_bind=bind(s->fd, (struct sockaddr *)&listen_addr6,sizeof(struct sockaddr_in6 ));
			}else	{
				ret_bind=bind(s->fd, (struct sockaddr *)&listen_addr,sizeof(struct sockaddr_in));
			}
			if (ret_bind < 0){
				ms_close(s->fd);
				ms_errGoto(fail, "bind faild %s(%d,%s)", s->url, errno, strerror(errno));
			}
			break;
		case 2://UDP组播输出，绑定IP地址
			ms_verbose("IP_MULTICAST_IF....");
			if_req.s_addr=inet_addr(localhostip); 
			if( setsockopt( s->fd, SOL_IP, IP_MULTICAST_IF, &if_req, sizeof(struct in_addr) ) < 0 ){ 
				ms_close(s->fd);
				ms_errGoto(fail, "[%s]fd setsockopt(IP_MULTICAST_IF) %s(%d,%s)", s->url, localhostip, errno, strerror(errno)); 
			} 
			break;
	}
	return 0;	
fail:
	return -1;
}
static ms_s08 udp_innerapi_udplite_coverage(URLContext *s)
{
	UDPContext * udp_ctt=s->priv_data;
	udp_ctt->udplite_coverage = UDP_HEADER_SIZE;
#if defined(OS_ANDROID)||defined(OS_LINUX_SOC)
	if (setsockopt (s->fd, IPPROTO_UDPLITE, UDPLITE_SEND_CSCOV, &(udp_ctt->udplite_coverage), sizeof(udp_ctt->udplite_coverage)) != 0){
		ms_errRet(-1, "fd setsockopt(UDPLITE_SEND_CSCOV) %s(%d,%s)", s->url, errno, strerror(errno));
	}
	if (setsockopt (s->fd, IPPROTO_UDPLITE, UDPLITE_RECV_CSCOV, &(udp_ctt->udplite_coverage), sizeof(udp_ctt->udplite_coverage)) != 0){
		ms_errRet(-1, "fd setsockopt(UDPLITE_RECV_CSCOV) %s(%d,%s)", s->url, errno, strerror(errno));
	}
#endif
	return 0;	
}
static ms_void udp_innerapi_epollInit(URLContext *s)
{
	UDPContext * udp_ctt=s->priv_data;
	udp_ctt->msepoll_fd=msepoll_open(2);
	if(s->flags&FLAG_WRITE){
		udp_ctt->msepoll_opt=msepollout;
	}else if(s->flags&FLAG_READ_WRITE){
		udp_ctt->msepoll_opt=(msepollout|msepollin);
	}else{
		udp_ctt->msepoll_opt=msepollin;
	}
	udp_ctt->msepoll_opt|=(msepollerr|msepollhup);
	msepoll_add(udp_ctt->msepoll_fd,s->fd,udp_ctt->msepoll_opt);

}
static int udp_open(URLContext *s)
{
	ms_bufcheckRet(-1, s);
	if(1==s->is_connected){
		return 0;
	}
/*初始化OPT变量*/	
	udp_optinit(s);
/*申请udp_ctt变量*/		
	UDPContext * udp_ctt=s->priv_data=(void *)ms_mallocDes(sizeof(UDPContext),"UDPContext",s->url);
	if(ms_null==s->priv_data){
		ms_errGoto(fail, "ms_malloc UDPContext faild(%d,%s)",errno,strerror(errno));
	}
/*解析URL地址*/
	msptc_api_urlSplit_urlctt( s);

/*IPV6时获取真实的IP地址*/
	if(udp_innerapi_getHostname(s)<0){
		ms_errGoto(fail, "udp_innerapi_getHostname failed");
	}
/*获取是否组播地址*/
	udp_ctt->is_multicast = msnet_api_isMulticast(s->url_info.hostname);
	ms_verbose("===========flags:[%d] url:[%s] s->is_multicast:[%d],client_prot:%d,flag_useipv6:%d",
		s->flags,s->url,udp_ctt->is_multicast,s->opt.udp.client_prot,s->url_info.flag_useipv6);
/*创建一个sockct*/
	if (udp_innerapi_createSocket(s) < 0){
		ms_errGoto(fail,  "socket %s failed(%d,%s)", s->url,errno,strerror(errno));
	}
/*获取绑定的IP地址，用于输入或输出绑定*/	
	ms_verbose("Get localhostip");
	int glocalipRet=0;
	char *localhostip=udp_innerapi_getLocalIp(s,&glocalipRet);
	if(glocalipRet<0){
		ms_errGoto(fail, "udp_innerapi_getLocalIp failed");
	}
/*设置发生或接收对端的信息*/	
	/*Init add info*/
	ms_verbose("set peer_info");
	udp_innerapi_setPeerInfo(s);
/*修改接收或发送缓存大小*/
	if(udp_innerapi_setBufSize(s)<0){
		ms_errGoto(fail, "udp_innerapi_setBufSize failed(%s)", s->url);
	}
	if(!(s->flags&FLAG_WRITE)	){	
		//绑定输入网卡:应对多个网卡都可以接收相同流的情况	
		if(ms_null!=s->ifcIn&& ms_buflen(s->ifcIn)){
			ms_verbose("msnet_api_bindIface:%s",s->ifcIn);
			if(msnet_api_bindIface(s->fd,s->ifcIn)<0){
				ms_errGoto(fail, "msnet_api_bindIface failed %s(%d,%s) ", s->url, errno, strerror(errno));
			}
		}
		//设置非阻塞或阻塞IO
		if (udp_innerapi_setIO(s) < 0){
			ms_errGoto(fail, "udp_innerapi_setIO failed");
		}
		//发送IGMP报文
		ms_verbose("set multicastgroup options");
		if(udp_ctt->is_multicast){
			if(udp_innerapi_igmp( s)<0){
				ms_errGoto(fail, "udp_innerapi_igmp failed(%s)", s->url);
			}
		}
	}else{
		//设置UDP的生存时间
		int mcast_ttl=64;	//router transmit node number
		if (setsockopt(s->fd,IPPROTO_IP,IP_MULTICAST_TTL,&mcast_ttl,sizeof(mcast_ttl))!= 0){
			ms_close(s->fd);
			ms_errGoto(fail, "setsockopt(IP_MULTICAST_TTL) %s(%d,%s)", s->url,errno,strerror(errno));
		}

		int loop=1;	//0-disable;1-enable
		if (setsockopt(s->fd,IPPROTO_IP,IP_MULTICAST_LOOP,&loop,sizeof(loop))!= 0){
			ms_close(s->fd);
			ms_errGoto(fail, "setsockopt(IP_MULTICAST_LOOP) %s(%d,%s)", s->url,errno,strerror(errno));
		}
		if(s->flags&FLAG_BROADCAST){
			int broadcast=1;
			if (setsockopt (s->fd, SOL_SOCKET, SO_BROADCAST, &(broadcast), sizeof(broadcast)) != 0){
				ms_close(s->fd);
				ms_errGoto(fail, "setsockopt(SO_BROADCAST) %s(%d,%s)", s->url,errno,strerror(errno));
			}
		}
	}

/*设置udplite_coverage*/	
	if (ms_strncmp_seq(s->url_info.ptcName,"udplite")) {
		if(udp_innerapi_udplite_coverage(s)<0){
			ms_errGoto(fail, "udp_innerapi_udplite_coverage failed(%s)", s->url);
		}
	}
/*绑定输入或输出IP信息*/		
	if(udp_innerapi_bind(s, localhostip)<0){
		ms_errGoto(fail, "udp_innerapi_bind failed(%s)", s->url);
	}
/*创建并添加epoll信息*/		
	udp_innerapi_epollInit(s);
	s->is_connected=1;	
	return 0;
 fail:
	if (ms_null!=udp_ctt){
		if(udp_ctt->msepoll_fd>0){
			msepoll_close(&udp_ctt->msepoll_fd);
		}
		if(udp_ctt->is_multicast &&!(s->flags&FLAG_WRITE)){
			igmp_api_leave(s);
		}
		ms_free(udp_ctt);
		udp_ctt=ms_null;
	}
	if(s->fd>0){
		ms_close(s->fd);
		msnet_api_deinit();
		s->fd=-1;
	}
	s->is_connected=0;
	return -1;
}
static int udp_read(URLContext *s ,  unsigned char *buf, int size)
{
	if(0==s->is_connected){
		return -1;
	}
  	UDPContext *udp_ctt =s->priv_data;	
	int ret;	
	if (0==s->opt.udp.block	
		&&s->opt.udp.rw_timeout>0){
		MSEPOLLRet ret_epoll=msepoll_waitfd_timeout_noopt(udp_ctt->msepoll_fd, s->fd, 0, s->opt.udp.rw_timeout);
		if (ret_epoll < 0){
			if(msepoll_ret_err_inner==ret_epoll){ms_error("[%s]msepoll error: %d %s",s->url,errno, strerror(errno));}
			return ret_epoll;
		}
	}
	if(s->flags&FLAG_READ_WRITE){
		if(s->url_info.flag_useipv6){
			ret=recvfrom(s->fd, buf, size, 0 , (struct sockaddr *)&s->peer_info.sockaddrs6, &s->peer_info.len);
		}else{
			ret=recvfrom(s->fd, buf, size, 0 , (struct sockaddr *)&s->peer_info.sockaddrs, &s->peer_info.len);
		}
		if( ret ==-1){
			if(s->url_info.flag_useipv6){
				ms_byte ipaddrbuf[256]={0};
				msnet_api_getIpaddr((struct sockaddr *)&s->peer_info.sockaddrs6,ipaddrbuf);
				ms_errRet(-1, "recvfrom(%s) net_error:  %d %s", ipaddrbuf,errno, strerror(errno));
			}else{
				ms_errRet(-1, "recvfrom(%s) net_error:  %d %s", inet_ntoa(s->peer_info.sockaddrs.sin_addr),errno, strerror(errno));
			}
		}else{
			//ms_verbose("[%s]recvfrom msg from %s,msg len is %d",s->url,inet_ntoa(s->peer_info.sockaddrs.sin_addr), ret);
			if(0==s->peer_info.has_info){
				if(s->url_info.flag_useipv6){
					ms_byte ipaddrbuf[256]={0};
					msnet_api_getIpaddr((struct sockaddr *)&s->peer_info.sockaddrs6,ipaddrbuf);
					ms_debug("get peer info,addr:%s(%d)",ipaddrbuf,s->peer_info.len);
				}else{
					ms_debug("get peer info,addr:%s(%d)",inet_ntoa(s->peer_info.sockaddrs.sin_addr),s->peer_info.len);
				}
			}
			s->peer_info.has_info=1;
		}
	}else{	
		ret = recv(s->fd, buf, size, 0);	
		if (ret<0){
			if (0==s->opt.udp.block	){
				ms_errRet(-1, "recv net_error:  %d %s (%s)",errno, strerror(errno), s->url);
			}else{
				return -1;
			}
		}else if(0==ret){
			ms_debug("No data to recv");
		}
	}
	return ret;

}
static int udp_write(URLContext *s, unsigned char *buf, int size)
{
	if(0==s->is_connected){
		return -1;
	}
	int ret;
	UDPContext * udp_ctt=s->priv_data;
	/* BEGIN: Added by su.gao, 2017/5/19 ,drop too long data*/
	if(UDP_TX_BUFSIZE<size){
		ms_waring("Drop %d byte,too long:(%s)",size, s->url);
		return 0;
	}
	/* END:   Added by su.gao, 2017/5/19 */

	/*打开发送的msepoll_waitfd_timeout_noopt检测会导致CPU资源的增加（AARCH64平台），因此关闭*/
#if 0	
	if (0==s->opt.udp.block	
		&&s->opt.udp.rw_timeout>0) {
#ifndef SOC_PLATFORM_RK3399
		MSEPOLLRet ret_epoll=msepoll_waitfd_timeout_noopt(udp_ctt->msepoll_fd, s->fd, 1, s->opt.udp.rw_timeout);
		if (ret_epoll < 0){
			if(msepoll_ret_err_inner==ret_epoll){ms_error("[%s]msepoll error: %d %s",s->url,errno, strerror(errno));}
			return ret_epoll;
		}	
#endif
	}
#endif
	if(( (s->flags&FLAG_READ_WRITE)&& (s->peer_info.has_info==1) )||(s->flags& FLAG_WRITE)){
		//ms_info("sendto  %s (%d toip:%s)", buf,size,inet_ntoa(s->client_info.clientaddr.sin_addr));
		if(s->url_info.flag_useipv6){
			ret=sendto(s->fd, buf, size, 0 , (struct sockaddr *)&s->peer_info.sockaddrs6, s->peer_info.len);
		}else{
			ret=sendto(s->fd, buf, size, 0 , (struct sockaddr *)&s->peer_info.sockaddrs, s->peer_info.len);
		}
		if( ret ==-1){
			if( ms_true==udp_ctt->is_multicast&&(ms_null!=s->opt.udp.ifcOut)){
				ms_info( "errno:%d",errno);
				if( 101== errno ){
					// ipaddr chage
					ms_byte ipaddr[256]={0};
					msnet_api_getIfaceIp(s->opt.udp.ifcOut, ipaddr);
					if(ms_strncmp_snaeq(ipaddr, s->opt.udp.ipOut) 
						&& ms_strncmp_snaeq(ipaddr, "0.0.0.0")
						&& ms_strncmp_snaeq( s->opt.udp.ipOut, "0.0.0.0")){
						ms_errRet(ptc_io_ipChange, "[%s]interface(%s) ipaddr chage from %s to %s,sendto(%s) net_error:  %d %s",
							s->url,s->opt.udp.ifcOut,s->opt.udp.ipOut,ipaddr,
							inet_ntoa(s->peer_info.sockaddrs.sin_addr),errno, strerror(errno));
					}
					if( ms_strncmp_saeq(ipaddr, "0.0.0.0")||ms_strncmp_saeq( s->opt.udp.ipOut, "0.0.0.0")){
						ms_errRet(ptc_io_linkDown, "[%s]Maybe interface(%s) link state change,ipaddr %s,opt.udp.ipOut:%s",
							s->url,s->opt.udp.ifcOut,ipaddr,s->opt.udp.ipOut);
					}
				}
			}
			ms_info( "errno:%d",errno);
			if(s->url_info.flag_useipv6){
				ms_byte ipaddrbuf[256]={0};
				msnet_api_getIpaddr((struct sockaddr *)&s->peer_info.sockaddrs6,ipaddrbuf);
				ms_errRet(0, "[%s]len:%d,sendto(%s) net_error:%d %s",s->url,size, ipaddrbuf,errno, strerror(errno));
			}else{
				ms_errRet(0, "[%s]len:%d,sendto(%s) net_error:%d %s",s->url,size,inet_ntoa(s->peer_info.sockaddrs.sin_addr),errno, strerror(errno));
			}
		}
	}else{
		ret = send(s->fd, buf, size, 0);
		if (ret<0){
			ms_errRet(0, "send net_error:  %d %s (%s)",errno, strerror(errno), s->url);
	    	}
	}
	return ret;
}

static int udp_close(URLContext *s)
{
	UDPContext * udp_ctt=s->priv_data;
	if(0==s->is_connected){
		return 0;
	}
	msepoll_del(udp_ctt->msepoll_fd,s->fd,udp_ctt->msepoll_opt);
	msepoll_close(&udp_ctt->msepoll_fd);
	if (ms_null!=udp_ctt){
		if(udp_ctt->is_multicast &&!(s->flags&FLAG_WRITE)){
			igmp_api_leave(s);
		}
		ms_free(udp_ctt);
		udp_ctt=ms_null;
	}
	if(s->fd>0){
		ms_close(s->fd);
		msnet_api_deinit();
		s->fd=-1;
	}
	s->is_connected=0;
	return 0;
}

URLProtocol ffurl_udp_protocol={
	.nameDes	= msptc_getPtcNameDes(msptc_udp),
    .url_open	= udp_open,
    .url_read	= udp_read,
    .url_write	= udp_write,
    .url_close	= udp_close,
    .flags		=STREAM_LIVE,
    .next	=ms_null,
};
URLProtocol ffurl_udplite_protocol={
	.nameDes	= msptc_getPtcNameDes(msptc_udplite),
    .url_open	= udp_open,
    .url_read	= udp_read,
    .url_write	= udp_write,
    .url_close	= udp_close,
    .flags		=STREAM_LIVE,
    .next	=ms_null,
};
#undef UDP_C
